package com.aotain.nyx.udpflood;

import java.util.HashMap;
import java.util.Map.Entry;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import com.aotain.nyx.flink.BasicTuple;


/**
 * 
     *@Title UDPFlood 检测 Reduce
     *@Description 
     *@Author Turk
     *@Since 2017年8月30日
     *@Version 1.1.0
 */
public class UDPFloodReduce implements ReduceFunction<Tuple2<String,HashMap<String,BasicTuple>>>{

	/**
	 *
	 */
	    
	private static final long serialVersionUID = 5052529257746200529L;

	@Override
	public Tuple2<String, HashMap<String,BasicTuple>> reduce(Tuple2<String, HashMap<String,BasicTuple>> value1,
			Tuple2<String, HashMap<String,BasicTuple>> value2) throws Exception {
		
		HashMap<String,BasicTuple> mapTuple = new HashMap<String,BasicTuple>();
		
		
		if(value1.f1 != null && value2.f1 != null) {
			HashMap<String,BasicTuple> map1 = value1.f1;
			HashMap<String,BasicTuple> map2 = value2.f1;
			for(Entry<String, BasicTuple> entry : map1.entrySet()) {
				BasicTuple tuple1 = map1.get(entry.getKey());
				if(map2.containsKey(entry.getKey())) {
					BasicTuple tuple2 = map2.get(entry.getKey());
					tuple1.setUpStreamOctets(tuple1.getUpStreamOctets() + tuple2.getUpStreamOctets());
					tuple1.setUpStreamPacket(tuple1.getUpStreamPacket() + tuple2.getUpStreamPacket());
					mapTuple.put(entry.getKey(), tuple1);
				} else {
					mapTuple.put(entry.getKey(), tuple1);
				}
			}
		} else {
			System.out.println("reduce arg is null");
		}
		
		return new Tuple2<String, HashMap<String,BasicTuple>> (value1.f0,mapTuple);
		    
	}

}

    